package com.liuqi.knet.remoting.exchange.support;

import com.liuqi.knet.Constants;
import com.liuqi.knet.remoting.Channel;
import com.liuqi.knet.remoting.RemotingException;
import com.liuqi.knet.remoting.exchange.Request;
import com.liuqi.knet.remoting.exchange.Response;
import com.liuqi.knet.remoting.exchange.ResponseFuture;
import com.liuqi.loggers.api.Logger;
import com.liuqi.loggers.facade.LoggersFacade;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 缓存所有请求的ResponseFuture实例, 通过此实例获取响应结果
 *
 * @author liuqi
 * @date 2024/12/17 15:26
 **/
public class DefaultResponseFuture implements ResponseFuture {

    private static final Logger logger = LoggersFacade.getLogger(DefaultResponseFuture.class);

    /**
     * 缓存所有请求
     */
    private static final Map<Long, DefaultResponseFuture> FUTURES = new ConcurrentHashMap<Long, DefaultResponseFuture>();

    static {
        Thread th = new Thread(new ResponseFutureTimeoutScan(), Constants.responseFutureTimeoutScanThreadName);
        th.setDaemon(true);
        th.start();
    }

    private final long id;
    private final Channel channel;
    private final Request request;
    private final int timeout;

    private final Lock lock = new ReentrantLock();
    private final Condition done = lock.newCondition();
    private final long start = System.currentTimeMillis();

    private volatile Response response;

    private volatile boolean cancelled;

    public DefaultResponseFuture(Channel channel, Request request, int timeout) {
        this.channel = channel;
        this.request = request;
        this.id = request.getId();
        this.timeout = timeout;

        // cache
        FUTURES.put(id, this);
    }

    /**
     * 异步结果回调
     *
     * @param response
     */
    public static void received(Response response) {
        DefaultResponseFuture future = FUTURES.remove(response.getId());
        if (future != null) {
            future.doReceived(response);
        }
    }

    public static boolean exist(long id) {
        return FUTURES.containsKey(id);
    }

    @Override
    public void cancel() {
        if (isDone()) {
            throw new RuntimeException("Response 已完成, 不能进行中断. request: " + request + ", channel: " + channel);
        }

        Response resp = new Response(id);
        resp.setErrorMessage("请求被取消");
        response = resp;
        cancelled = true;
        FUTURES.remove(id);
    }

    @Override
    public boolean isCancelled() {
        return cancelled;
    }

    @Override
    public Response get() throws RemotingException {
        return get(timeout, TimeUnit.MILLISECONDS);
    }

    @Override
    public Response get(int timeout, TimeUnit timeUnit) throws RemotingException {
        if (!isDone()) {
            long start = System.currentTimeMillis();
            lock.lock();
            try {
                while (!isDone()) {
                    done.await(timeout, TimeUnit.MILLISECONDS);
                    if (isDone() || System.currentTimeMillis() - start > timeout) {
                        break;
                    }
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
            if (!isDone()) {
                timeoutResponse(this);
            }
        }
        return response;
    }

    @Override
    public boolean isDone() {
        return response != null;
    }

    public long getId() {
        return id;
    }

    public Request getRequest() {
        return request;
    }

    private void doReceived(Response res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
    }

    private static void timeoutResponse(DefaultResponseFuture future) {
        Response errResp = new Response(future.getId());
        errResp.setState(Response.RESPONSE_TIMEOUT);
        errResp.setErrorMessage(future.getTimeoutMessage());

        //  handle response.
        DefaultResponseFuture.received(errResp);
    }

    private String getTimeoutMessage() {
        return "等待服务器端响应超时: " +
                "消息发送时间: " + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date(start))) + ", " +
                "超时时间: [" + timeout + "]ms, " +
                "request: " + request + ", " +
                "channel: " + channel;
    }

    private static class ResponseFutureTimeoutScan implements Runnable {
        @Override
        public void run() {
            logger.info("启动线程 - 定时扫描等待响应超时的请求");
            while (true) {
                try {
                    for (DefaultResponseFuture future : FUTURES.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }

                        // 超时
                        if (System.currentTimeMillis() - future.start > future.timeout) {
                            timeoutResponse(future);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable t) {
                    logger.error("定时扫描读取响应超时的请求, 出现异常.", t);
                }
            }
        }
    }
}
